package com.sunzm.flink.datastream.scala.typeinfomation

import com.sunzm.flink.utils.FlinkEnvUtils
import org.apache.commons.lang3.StringUtils
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
import org.apache.flink.api.scala.typeutils.{Types}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.createTypeInformation
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 *
 * ${DESCRIPTION}
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-12-11 14:48
 */
object ScalaTypeInfomationDemo {

  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName)

  def main(args: Array[String]): Unit = {
    val env = FlinkEnvUtils.getStreamExecutionEnvironment(true)

    val dataStream = env.socketTextStream("82.156.210.70", 9999)
      .flatMap((line: String, collector: Collector[String]) => {

        try {
          if (StringUtils.isNotBlank(line)) {
            //截取字符串
            val words: Array[String] = line.split(",");

            words.foreach(word => {
              collector.collect(word)
            })

          }
        } catch {
          case e: Throwable => {
            logger.error(s"数据格式异常: ${line}", e)
          }
        }

      })(Types.STRING)
      .keyBy(word => word)

    val resultDS = dataStream.flatMap(new RichFlatMapFunction[String, (String, Long)] {

      /**
       * 记录每个单词次数的状态
       */
      private var wordCountState: ValueState[Long] = _


      override def open(parameters: Configuration): Unit = {

        wordCountState = getRuntimeContext.getState(new ValueStateDescriptor[Long](
          "scalaWordCountState",
          createTypeInformation[Long]
        ))

      }

      override def flatMap(word: String, out: Collector[(String, Long)]): Unit = {

        var stateWordCount = wordCountState.value()

        if (stateWordCount == null) stateWordCount = 0L

        val newWordCount = stateWordCount + 1

        //更新状态数据
        wordCountState.update(newWordCount)

        //输出结果
        out.collect((word, newWordCount))
      }
    })(createTypeInformation[(String, Long)])

    resultDS.print("单词次数[scala]")

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
